#include <signal.h>
#ifdef __linux__
#include <pthread.h>
#endif
#include <list>
#include <queue>
#include <rise/common/MutablePtr.h>
#include <rise/common/Log.h>
#include <rise/common/ExceptionTemplate.h>
#include <rise/threading/CriticalSection.h>
#include <rise/socket/ServerSocket.h>
#include <rise/tools/StackTracer.h>
#include "MessageQueueManager.h"
#include "ClientWorker.h"
#include "MQServer.h"

namespace smq
{
  struct MQServer::MQServerImpl
  {
    typedef rise::CMutablePtr<ClientWorker> PClientWorker;
    typedef std::list<PClientWorker> ClientWorkerList;

    MQServerImpl():
      bStopFlag(false)
    {
    }

    static void OnSignal(int nSignal)
    {
      rise::LogEntry();
      rise::LogNotice() << "Signal: " << nSignal;
      switch (nSignal)
      {
#ifndef WIN32
      case SIGCHLD:
        {
          MQServer::Inst().m_pImpl->OnChildExit();
          break;
        }
#endif
        case SIGINT:
        {
          MQServer::Inst().Stop();
          break;
        }
        case SIGSEGV:
        {
          std::string sTracedStack;
          rise::tools::CStackTracer::StackTraceStr(sTracedStack);
          rise::LogError() << "Segmentation fault.\nTraced stack:\n" << sTracedStack;
          exit(1);
          break;
        }
      }
    }

    void OnChildExit()
    {
      rise::LogEntry();
      rise::threading::CScopedCriticalSection tScopedLockMessage(MessageQueueManager::Inst().GetLock()); // lock message handling
      rise::threading::CScopedCriticalSection tScopedLock(tLock);
      for (ClientWorkerList::iterator itWorker = lsClientWorkers.begin(); itWorker != lsClientWorkers.end();)
      {
        if ((*itWorker)->IsFinished())
        {
          rise::LogDebug() << (*itWorker)->GetType() << " \"" << (*itWorker)->GetName() << "\" has ended. waiting...";
          // waiting for detached thread
          (*itWorker)->Join();
          lsClientWorkers.erase(itWorker++);
          rise::LogDebug() << "done";
        }
        else
        {
          ++itWorker;
        }
      }
    }

    void Start(const std::string& sServerHost, unsigned short ushServerPort, int nMaxQueue)
    {
      rise::LogEntry();
      bool bReadyToAccept = false;

#ifndef WIN32
      signal(SIGCHLD, OnSignal);
#endif
      signal(SIGINT, OnSignal);
      signal(SIGSEGV, OnSignal);

      unsigned long ulAddress = ((sServerHost == "") || (sServerHost == "all"))
        ? INADDR_ANY : inet_addr(sServerHost.c_str());

      tServerSocket.Create(ushServerPort, ulAddress, rise::sockets::CSocket::ST_STREAM, nMaxQueue);

      rise::LogNotice() << " _____ STARTING _____ ";

      bStopFlag = false;
      while (!bStopFlag)
      {
        PClientWorker pClientWorker(new ClientWorker);
        while (!bStopFlag)
        {
          bReadyToAccept = Select();
          if (bStopFlag)
          {
            break;
          }

          if (bReadyToAccept && tServerSocket.Accept(pClientWorker->GetSocket()))
          {
            if (bStopFlag)
            {
              break;
            }

            {
              rise::threading::CScopedCriticalSection tScopedLock(tLock);
              lsClientWorkers.push_back(pClientWorker);
            }
            pClientWorker->Start();
            break;
          }
        }
      }

      rise::LogNotice() << " _____ STOPPING  _____ ";
    }

    bool Select()
    {
      bool bReadyToAccept = false;
      rise::SOCKET hSockMax = tServerSocket.GetHandle();
      fd_set stRead;

      FD_ZERO(&stRead);

      FD_SET(tServerSocket.GetHandle(), &stRead);

      int nRes = select(hSockMax + 1, &stRead, NULL, NULL, NULL);

      if (nRes > 0) // some sockets are signalled
      {
        if (FD_ISSET(tServerSocket.GetHandle(), &stRead))
        {
          rise::LogDebug2() << "Server is Ready to read (accept, closed/reset/terminated)";
          bReadyToAccept = true;
        }
      }

      return bReadyToAccept;
    }

    rise::sockets::CServerSocket tServerSocket;
    rise::threading::CCriticalSection tLock;
    ClientWorkerList lsClientWorkers;
    bool bStopFlag;
  };

  MQServer& MQServer::Inst()
  {
    if (!m_pInst)
    {
      m_pInst = new MQServer;
    }

    return *m_pInst;
  }

  void MQServer::FreeInst()
  {
    if (m_pInst)
    {
      delete m_pInst;
      m_pInst = NULL;
    }
  }

  void MQServer::Start(const std::string& sServerHost /*= "127.0.0.1"*/, unsigned short ushServerPort /*= 8768*/, int nMaxQueue /*= 10*/)
  {
    m_pImpl->Start(sServerHost, ushServerPort, nMaxQueue);
  }

  void MQServer::Stop()
  {
    m_pImpl->bStopFlag = true;
#ifdef WIN32
    m_pImpl->tServerSocket.Close();
#endif
  }

  MQServer::MQServer()
  {
    m_pImpl = new MQServerImpl;
  }

  MQServer::~MQServer()
  {
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  MQServer* MQServer::m_pInst = NULL;
}
